package site.teamo.biu.flink.function.source;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import site.teamo.biu.flink.sdk.annotation.function.BFSourceFunction;
import site.teamo.biu.flink.sdk.function.AbstractSourceFunction;
import site.teamo.biu.flink.sdk.job.JobContext;

import java.util.Random;

/**
 * @author haocongshun
 * @date 2023/07/25 15:20:07
 */
@BFSourceFunction(
        out = Integer.class
)
@Slf4j
public class RandomNumberSource extends AbstractSourceFunction<Integer> {

    private final int min;
    private final int max;
    private final Random random;

    private volatile boolean flag = true;

    public RandomNumberSource(String name, JobContext context) {
        super(name, context);
        this.min = context.getConfig(this.name).getInteger("min");
        this.max = context.getConfig(this.name).getInteger("max");
        this.random = new Random();
    }

    @Override
    public void run(SourceContext<Integer> sourceContext) throws Exception {
        while (flag) {
            sourceContext.collect(random.nextInt(max - min) + min);
            ThreadUtil.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}
